package com.develophelper.aliyun.rocketmq.core;

import com.aliyun.openservices.ons.api.*;
import com.aliyun.openservices.ons.api.order.OrderProducer;
import com.develophelper.aliyun.rocketmq.autoconfigure.RocketMQProperties;
import com.develophelper.aliyun.rocketmq.support.RocketMQMessageConverter;
import com.develophelper.aliyun.rocketmq.support.RocketMQUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.core.AbstractMessageSendingTemplate;
import org.springframework.messaging.core.MessagePostProcessor;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.MimeTypeUtils;

import java.util.Map;
import java.util.Objects;
import java.util.Properties;

/**
 * @author snaildev
 * @date 2021/05/17
 **/
public class RocketMQTemplate extends AbstractMessageSendingTemplate<String> implements InitializingBean, DisposableBean {
    private static final Logger log = LoggerFactory.getLogger(RocketMQTemplate.class);

    private RocketMQProperties mqProperties;

    private Producer producer;

    private OrderProducer orderProducer;

//    private TransactionProducer transactionProducer;

    private String charset = "UTF-8";

    private RocketMQMessageConverter rocketMQMessageConverter = new RocketMQMessageConverter();

    public RocketMQTemplate(RocketMQProperties mqProperties) {
        this.mqProperties = mqProperties;
        Properties properties = getProperties();
        producer = ONSFactory.createProducer(properties);
        orderProducer = ONSFactory.createOrderProducer(properties);
//        transactionProducer = ONSFactory.createTransactionProducer(properties);
    }

    private Properties getProperties() {
        Properties properties = new Properties();
        properties.setProperty(PropertyKeyConst.NAMESRV_ADDR, this.mqProperties.getNameServer());
        properties.setProperty(PropertyKeyConst.GROUP_ID, this.mqProperties.getGroupId());
        properties.setProperty(PropertyKeyConst.AccessKey, this.mqProperties.getAccessKey());
        properties.setProperty(PropertyKeyConst.SecretKey, this.mqProperties.getSecretKey());
        return properties;
    }

    /**
     * <p> Send message in synchronous mode. This method returns only when the sending procedure totally completes.
     * Reliable synchronous transmission is used in extensive scenes, such as important notification messages, SMS
     * notification, SMS marketing system, etc.. </p>
     * <p>
     *
     * @param destination formats: `topicName:tags`
     * @param message     {@link org.springframework.messaging.Message}
     * @return {@link SendResult}
     */
    public SendResult syncSend(String destination, Message<?> message) {
        return syncSend(destination, message, 0);
    }

    /**
     * Same to {@link #syncSend(String, Message)} with send timeout specified in addition.
     *
     * @param destination formats: `topicName:tags`
     * @param message     {@link org.springframework.messaging.Message}
     * @param delayTime   time for the delay message
     * @return {@link SendResult}
     */
    public SendResult syncSend(String destination, Message<?> message, long delayTime) {
        if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
            log.error("syncSend failed. destination:{}, message is null ", destination);
            throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
        }
        try {
            long now = System.currentTimeMillis();
            com.aliyun.openservices.ons.api.Message rocketMsg = this.createRocketMqMessage(destination, message);
            if (delayTime > 0) {
                rocketMsg.setStartDeliverTime(delayTime);
            }
            SendResult sendResult = producer.send(rocketMsg);
            long costTime = System.currentTimeMillis() - now;
            if (log.isDebugEnabled()) {
                log.debug("send message cost: {} ms, msgId:{}", costTime, sendResult.getMessageId());
            }
            return sendResult;
        } catch (Exception e) {
            log.error("syncSend failed. destination:{}, message:{} ", destination, message);
            throw new MessagingException(e.getMessage(), e);
        }
    }

    /**
     * Same to {@link #syncSend(String, Message)}.
     *
     * @param destination formats: `topicName:tags`
     * @param payload     the Object to use as payload
     * @return {@link SendResult}
     */
    public SendResult syncSend(String destination, Object payload) {
        return syncSend(destination, payload, 0);
    }

    /**
     * Same to {@link #syncSend(String, Object)} with send timeout specified in addition.
     *
     * @param destination formats: `topicName:tags`
     * @param payload     the Object to use as payload
     * @param delayTime   time for the delay message
     * @return {@link SendResult}
     */
    public SendResult syncSend(String destination, Object payload, long delayTime) {
        Message<?> message = MessageBuilder.withPayload(payload).build();
        return syncSend(destination, message, delayTime);
    }

    /**
     * Same to {@link #syncSendOrderly(String, Message, String)} with send timeout specified in addition.
     *
     * @param destination formats: `topicName:tags`
     * @param message     {@link org.springframework.messaging.Message}
     * @param hashKey     use this key to select queue. for example: orderId, productId ...
     * @return {@link SendResult}
     */
    public SendResult syncSendOrderly(String destination, Message<?> message, String hashKey) {
        if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
            log.error("syncSendOrderly failed. destination:{}, message is null ", destination);
            throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
        }
        try {
            long now = System.currentTimeMillis();
            com.aliyun.openservices.ons.api.Message rocketMsg = this.createRocketMqMessage(destination, message);
            SendResult sendResult = orderProducer.send(rocketMsg, hashKey);
            long costTime = System.currentTimeMillis() - now;
            if (log.isDebugEnabled()) {
                log.debug("send message cost: {} ms, msgId:{}", costTime, sendResult.getMessageId());
            }
            return sendResult;
        } catch (Exception e) {
            log.error("syncSendOrderly failed. destination:{}, message:{} ", destination, message);
            throw new MessagingException(e.getMessage(), e);
        }
    }

    /**
     * Same to {@link #syncSend(String, Object)} with send orderly with hashKey by specified.
     *
     * @param destination formats: `topicName:tags`
     * @param payload     the Object to use as payload
     * @param hashKey     use this key to select queue. for example: orderId, productId ...
     * @return {@link SendResult}
     */
    public SendResult syncSendOrderly(String destination, Object payload, String hashKey) {
        return syncSendOrderly(destination, payload, hashKey);
    }

    /**
     * Same to {@link #syncSendOrderly(String, Object, String)} with send timeout specified in addition.
     *
     * @param destination formats: `topicName:tags`
     * @param payload     the Object to use as payload
     * @param hashKey     use this key to select queue. for example: orderId, productId ...
     * @param timeout     send timeout with millis
     * @return {@link SendResult}
     */
    public SendResult syncSendOrderly(String destination, Object payload, String hashKey, long timeout) {
        Message<?> message = MessageBuilder.withPayload(payload).build();
        return syncSendOrderly(destination, message, hashKey, timeout);
    }

    /**
     * Same to {@link #asyncSend(String, Message, SendCallback)} with send timeout and delay level specified in
     * addition.
     *
     * @param destination  formats: `topicName:tags`
     * @param message      {@link org.springframework.messaging.Message}
     * @param sendCallback {@link SendCallback}
     * @param delayTime    send timeout with millis
     */
    public void asyncSend(String destination, Message<?> message, SendCallback sendCallback, long delayTime) {
        if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
            log.error("asyncSend failed. destination:{}, message is null ", destination);
            throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
        }
        try {
            com.aliyun.openservices.ons.api.Message rocketMsg = this.createRocketMqMessage(destination, message);
            if (delayTime > 0) {
                rocketMsg.setStartDeliverTime(delayTime);
            }
            producer.sendAsync(rocketMsg, sendCallback);
        } catch (Exception e) {
            log.info("asyncSend failed. destination:{}, message:{} ", destination, message);
            throw new MessagingException(e.getMessage(), e);
        }
    }

    /**
     * <p> Send message to broker asynchronously. asynchronous transmission is generally used in response time
     * sensitive business scenarios. </p>
     * <p>
     * This method returns immediately. On sending completion, <code>sendCallback</code> will be executed.
     * <p>
     *
     * @param destination  formats: `topicName:tags`
     * @param message      {@link org.springframework.messaging.Message}
     * @param sendCallback {@link SendCallback}
     */
    public void asyncSend(String destination, Message<?> message, SendCallback sendCallback) {
        asyncSend(destination, message, sendCallback, 0);
    }

    /**
     * Same to {@link #asyncSend(String, Object, SendCallback)} with send timeout specified in addition.
     *
     * @param destination  formats: `topicName:tags`
     * @param payload      the Object to use as payload
     * @param sendCallback {@link SendCallback}
     */
    public void asyncSend(String destination, Object payload, SendCallback sendCallback) {
        Message<?> message = MessageBuilder.withPayload(payload).build();
        asyncSend(destination, message, sendCallback, 0);
    }

    /**
     * Similar to <a href="https://en.wikipedia.org/wiki/User_Datagram_Protocol">UDP</a>, this method won't wait for
     * acknowledgement from broker before return. Obviously, it has maximums throughput yet potentials of message loss.
     * <p>
     * One-way transmission is used for cases requiring moderate reliability, such as log collection.
     *
     * @param destination formats: `topicName:tags`
     * @param message     {@link org.springframework.messaging.Message}
     */
    public void sendOneWay(String destination, Message<?> message) {
        if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
            log.error("sendOneWay failed. destination:{}, message is null ", destination);
            throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
        }
        try {
            com.aliyun.openservices.ons.api.Message rocketMsg = this.createRocketMqMessage(destination, message);
            producer.sendOneway(rocketMsg);
        } catch (Exception e) {
            log.error("sendOneWay failed. destination:{}, message:{} ", destination, message);
            throw new MessagingException(e.getMessage(), e);
        }
    }

    /**
     * Same to {@link #sendOneWay(String, Message)}
     *
     * @param destination formats: `topicName:tags`
     * @param payload     the Object to use as payload
     */
    public void sendOneWay(String destination, Object payload) {
        Message<?> message = MessageBuilder.withPayload(payload).build();
        sendOneWay(destination, message);
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        if (Objects.nonNull(producer)) {
            producer.start();
        }

        if (Objects.nonNull(orderProducer)) {
            orderProducer.start();
        }

//        if (Objects.nonNull(transactionProducer)) {
//            transactionProducer.start();
//        }
    }

    @Override
    protected void doSend(String destination, Message<?> message) {
        SendResult sendResult = syncSend(destination, message);
        if (log.isDebugEnabled()) {
            log.debug("send message to `{}` finished. result:{}", destination, sendResult);
        }
    }

    @Override
    protected Message<?> doConvert(Object payload, Map<String, Object> headers, MessagePostProcessor postProcessor) {
        Message<?> message = super.doConvert(payload, headers, postProcessor);
        MessageBuilder<?> builder = MessageBuilder.fromMessage(message);
        builder.setHeaderIfAbsent(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN);
        return builder.build();
    }

    @Override
    public void destroy() {
        if (Objects.nonNull(producer)) {
            producer.shutdown();
        }

        if (Objects.nonNull(orderProducer)) {
            orderProducer.shutdown();
        }

//        if (Objects.nonNull(transactionProducer)) {
//            transactionProducer.shutdown();
//        }
    }

//    /**
//     * Send Spring Message in Transaction
//     *
//     * @param destination destination formats: `topicName:tags`
//     * @param message     message {@link org.springframework.messaging.Message}
//     * @param arg         ext arg
//     * @return TransactionSendResult
//     * @throws MessagingException
//     */
//    public SendResult sendMessageInTransaction(final String destination,
//                                               final Message<?> message, final LocalTransactionExecuter executer, final Object arg) throws MessagingException {
//        com.aliyun.openservices.ons.api.Message rocketMsg = this.createRocketMqMessage(destination, message);
//        return transactionProducer.send(rocketMsg, executer, arg);
//    }

    private com.aliyun.openservices.ons.api.Message createRocketMqMessage(
            String destination, Message<?> message) {
        Message<?> msg = this.doConvert(message.getPayload(), message.getHeaders(), null);
        com.aliyun.openservices.ons.api.Message rocketMessage = RocketMQUtil.convertToRocketMessage(getMessageConverter(), charset,
                destination, msg);

        // add trace tag
        rocketMessage.putUserProperties(mqProperties.getTraceId(), MDC.get(mqProperties.getTraceId()));
        return rocketMessage;
    }
}
